package org.example
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider.CUSTOM_METRIC_FILES_REUSED.desc
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
object data1_SQL3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    //读取电影用户信息
    val schemaUser = StructType(Seq(
      StructField("id", IntegerType),
      StructField("gender", StringType),
      StructField("age", IntegerType),
      StructField("occupation", IntegerType),
      StructField("zipcode", StringType)
    ))
    val user = spark.read.option("sep","::").schema(schemaUser)
      .csv("src/main/resources/users.dat")
    user.show(5)
    //1.查询where/filter
    user.where("gender = 'F' and age = 18").show(5)
    user.filter("gender = 'F' and age = 18").show(5)
    //替换udf
    spark.udf.register("replace", (x:String) => {
      x match {
        case "M" => 0
        case "F" => 1
      }
    })
    user.selectExpr("id", "replace(gender) as sexual", "age").show(3)
    //col()apply()方法
    //user.select(user.col("zipcode")).collect()
    //2.排序
    //user.orderBy(desc("id")).show(5)
    user.orderBy(-user("id")).show(5)
    //user.sort(desc("id")).show(5)
    //3.分组groupBy
    user.groupBy("gender").count().show(5)
    //4.连接 join(DataFrame,"列名")
    //练习：读取评分表，并将用户表和评分表连接，查询18岁女生评分为5分的所有电影id
    val schemaRatings = StructType(Seq(
      StructField("user_id", IntegerType),
      StructField("movie_id", IntegerType),
      StructField("rating", IntegerType),
      StructField("timestamp", LongType),
    ))
      val ratings = spark.read.option("sep","::").schema(schemaRatings)
        .csv("src/main/resources/ratings.dat")
    val joinedDF = user.join(ratings, user("id") === ratings("user_id"))
    val result = joinedDF
      .filter("gender = 'F' and age = 18 and rating = 5")
      .select("movie_id")
      .distinct()
    result.show()
    sc.stop()
  }
}
